package com.craiditx.udtf;

import com.github.jcrfsuite.example.UDFTag;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.xssf.usermodel.XSSFRow;
import org.apache.poi.xssf.usermodel.XSSFSheet;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;

import java.io.File;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


public class merchant_ner extends GenericUDTF {
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        ArrayList<String> fieldNames = new ArrayList<String>();
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
        fieldNames.add("keyword");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldNames.add("business");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

    /**
     *将模型文件写入到临时目录
     */
    public void writeModel() throws Exception {
        InputStream stream = this.getClass().getResourceAsStream("/models92mc71t.crfsuite");
        byte[] buffer = new byte[8192];
        String filename="/tmp/models92mc71t.crfsuite";
        File file=new File(filename);
        if(!file.exists()){
            file.createNewFile();
        }
        FileOutputStream writer = new FileOutputStream(filename);
        int bytesRead = 0;
        while ((bytesRead = stream.read(buffer)) != -1) {
            writer.write(buffer, 0, bytesRead);
        }
        stream.close();
        writer.close();
    }

    /**
     *获取模型结果
     */
    public static Map<String, String> getModelMappingResult(String modelFile,String inputData) throws Exception {
        return UDFTag.getModelResult(modelFile, inputData);
    }

    /**
     * 读取excel中标签数据
     */
    public List<Map<String,String>> getExcel() throws Exception{
        InputStream inputStream=this.getClass().getResourceAsStream("/trade_mapping_setting.xlsx");
        XSSFWorkbook sheets = new XSSFWorkbook(inputStream);
        XSSFSheet sheetAt = sheets.getSheetAt(0);
        List<Map<String,String>> excelData=new ArrayList<>();
        for(int i=1;i<sheetAt.getPhysicalNumberOfRows();i++){
            XSSFRow row = sheetAt.getRow(i);
            Iterator<Cell> cellIterator = row.cellIterator();
            Map<String,String> cell_map=new HashMap<>();
            int count=1;
            while(cellIterator.hasNext()){
                Cell next = cellIterator.next();
                if(next.toString()==""){cell_map.put("lev"+count,"null");}
                else{cell_map.put("lev"+count,next.toString());}
                count++;
            }
            excelData.add(cell_map);
        }
        inputStream.close();
        return excelData;
    }

    /**
     * 剔除mapping映射数据中多余的 | 符号
     */
    public static String clearRegex(String regex){
        char c=regex.charAt(0);
        char end=regex.charAt(regex.length()-1);
        StringBuffer buffer=new StringBuffer();
        buffer.append(c);
        for(int i=1;i<regex.length()-1;i++){
            if(c!='|' || regex.charAt(i)!='|'){buffer.append(regex.charAt(i));c=regex.charAt(i);}
        }
        if(end != '|'){buffer.append(end);}
        return buffer.toString();
    }

    /**
     *正则匹配字符串
     */
    public static boolean contData(String data,String regex){
        Matcher matcher = Pattern.compile(regex).matcher(data);
        boolean flag=false;
        while(matcher.find()){
            flag=true;
            break;
        }
        return flag;
    }

    /**
     *关键字与标签的映射，入参为模型结果传入,正则为excel中的商户关键字与经营范围
     */
    public static List<Map<String,String>> mappingFeature(Map<String,String> mapFeature) throws Exception {
        List<Map<String, String>> excel = new merchant_ner().getExcel();
        List<Map<String,String>> mapping_r=new ArrayList<>();
        String text=mapFeature.get("KEYWORD")+mapFeature.get("BUSINESS");
        if(!mapFeature.isEmpty()){
            for (Map<String, String> map : excel) {
                String KEYWORD=map.get("lev4");
                String BUSINESS=map.get("lev5");
                String KEYWORD_MODEL=mapFeature.get("KEYWORD");
                String BUSINESS_MODEL=mapFeature.get("BUSINESS");
                Map<String,String> mapR=new HashMap<>();
                if(contData(BUSINESS_MODEL, clearRegex(BUSINESS))){
                    mapR.put("text",text);
                    mapR.put("lev1",map.get("lev1"));
                    mapR.put("lev2",map.get("lev2"));
                    mapR.put("lev3",map.get("lev3"));
                    mapping_r.add(mapR);
                    break;
                }else if(contData(KEYWORD_MODEL, clearRegex(KEYWORD))){
                    mapR.put("text",text);
                    mapR.put("lev1",map.get("lev1"));
                    mapR.put("lev2",map.get("lev2"));
                    mapR.put("lev3",map.get("lev3"));
                    mapping_r.add(mapR);
                    break;
                }
            }
        }
        Map<String,String> mapOther=new HashMap<>();
        mapOther.put("text",text);
        mapOther.put("lev1","null");
        mapOther.put("lev2","null");
        mapOther.put("lev3","null");
        if(mapping_r.size()<1){mapping_r.add(mapOther);}
        return mapping_r;
    }

    @Override
    public void process(Object[] args) throws HiveException {
        String input = args[0].toString();
        try {
            writeModel();
            Map<String, String> modelResult = getModelMappingResult("/tmp/models92mc71t.crfsuite", input);
            String[] resultSet = new String[2];
            resultSet[0] = modelResult.get("KEYWORD");
            resultSet[1] = modelResult.get("BUSINESS");
//            System.out.println(resultSet[0]+" "+resultSet[1]);
            forward(resultSet);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws HiveException {

    }

    public static void main(String[] args) throws HiveException {
        merchant_ner merchant_ner = new merchant_ner();
        String[] arr = {"良古艺术主题餐厅"};
        merchant_ner.process(arr);
    }
}
